package javaProduct;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

public class Product {
    private final kafka.javaapi.producer.Producer<Integer, String> producer;
    private final String topic;
    private final Properties props = new Properties();
    public Product(String topic)
    {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", "s60:9092,s2:9092,s5:9092");
        props.put("compression.codec","snappy");
        props.put( "key.serializer.class","kafka.serializer.StringEncoder");
        props.put("producer.type","async");
        producer = new kafka.javaapi.producer.Producer(new ProducerConfig(props));
        this.topic = topic;
    }

    public void send(String message) {
            producer.send(new KeyedMessage<Integer, String>(topic, message));

    }
}
